package com.lzl.producer;

import com.lzl.common.GateWayPo;
import com.lzl.common.http.SimpleResponse;
import com.lzl.producer.services.ProducerService;
import com.lzl.producer.task.ProducerTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

@Component
@Slf4j
public class Producer {

    ConcurrentHashMap<Long, GateWayPo> callBackWaitList = new ConcurrentHashMap<>(3000);

    ExecutorService dbInputExcutor = new ThreadPoolExecutor(3, 5, 1,
            TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1000));

    AtomicBoolean running = new AtomicBoolean(true);

    AtomicLong total = new AtomicLong(0);

    @Autowired
    ProducerService service;

    public boolean submit(GateWayPo po) {
        final int waiting = callBackWaitList.size();
        if (waiting > 5000) {
            log.info("请稍后再试");
            return false;
        }
        try {
            ProducerTask task = new ProducerTask(po, service);
            dbInputExcutor.submit(task);
            callBackWaitList.put(po.getId(), po);
            total.incrementAndGet();
            return true;
        } catch (Exception e) {
            log.error(e.getMessage());
            return false;
        }


    }

    public void start() {
        new ResponseThread().start();
    }

    class ResponseThread extends Thread {
        public void run(){
            log.info("开始查询反馈结果");
            while(running.get()){
                try {
                    List<Long> gids = waitListGids();
                    if (gids.size() <= 0) {
                        TimeUnit.SECONDS.sleep(1);
                        continue;
                    }
                    log.info("查询范围：" + gids);
                    List<GateWayPo> es =  service.getResponseList(gids);
                    log.info("查询结果：" + es);
                    for (GateWayPo po :
                            es) {
                        GateWayPo ex = callBackWaitList.get(po.getId());
                        ex.setResponse(po.getResponse());
                        ex.doneSignal();
                        callBackWaitList.remove(po.getId());
                        log.info("处理完毕");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 等待队列的gid集合
     * @return
     */
    private List<Long> waitListGids(){
        List<Long> gids = new ArrayList<>();
        for(GateWayPo po : callBackWaitList.values()) {
                gids.add(po.getId());
        }
        return gids;
    }


}
